package com.asiainfo.dacp.datastash.broker.facade.service;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import com.asiainfo.dacp.datastash.broker.core.BrokerException;
import com.asiainfo.dacp.datastash.broker.core.enums.UnitStepTypeEnum;
import com.asiainfo.dacp.datastash.broker.core.vo.StashUnit;
import com.asiainfo.dacp.datastash.broker.core.vo.StashUnitDefine;
import com.asiainfo.dacp.datastash.broker.core.vo.StashUnitProcess;
import com.asiainfo.dacp.datastash.broker.core.vo.StashUnitStep;
import com.asiainfo.dacp.datastash.broker.core.vo.StashUnitTask;
import com.asiainfo.dacp.datastash.broker.tracelog.StashTraceLogVo;
import com.asiainfo.dacp.datastash.broker.tracelog.StashTraceLogger;
import com.asiainfo.dacp.datastash.domain.unit.models.DataStashUnitStep;
import com.google.common.base.Preconditions;
import com.google.common.base.Strings;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;

/**
 * 编译unit的服务类，把unit编译成多个task，根据引擎，分片的混合规则，解析成多个task的依赖
 *
 * @author MeiKefu
 * @date 2017-12-13
 */
@Service
public class UnitCompileService {

    @Autowired
    private StashTraceLogger tracelog;
    private String clzName = this.getClass().getSimpleName();

    @Autowired
    private UnitShardingService unitShardingService;

    public StashUnitProcess compile(StashUnit stashUnit){
        StashUnitProcess stashUnitProcess = new StashUnitProcess(stashUnit.getUnitProcId(),stashUnit.getUnitCode(),stashUnit.getBatchNo());

        try {
            StashUnitDefine stashUnitDefine = stashUnit.getStashUnitDefine();
            stashUnitProcess.setUnitPriority(stashUnitDefine.getUnitPriority());

            List<DataStashUnitStep> rootUnitStepDomains = stashUnitDefine.getRootSteps();

            List<StashUnitTask> taskList = new ArrayList<>();

            for (DataStashUnitStep stepDomain: rootUnitStepDomains){

               /* if(!UnitStepTypeEnum.Schedule.value().equals(stepDomain.getStepDefine().getStepType())){
                    StashUnitTask curTask = new StashUnitTask(stashUnit.getUnitProcId(),stashUnit.getUnitCode(),stashUnit.getBatchNo());
                    taskList.add(curTask);
                    String curEngine = stepDomain.getStepDefine().getStepType();

                }*/
                //recursiveTask(stepDomain,null,null,taskList,stashUnit);
            }

            //stashUnitProcess.setTasks(taskList);
            List<StashUnitTask> tasks = buildTask(stashUnitDefine, stashUnit);
            stashUnitProcess.setTasks(tasks);
            

        } catch (BrokerException e) {
            e.printStackTrace();
        }

        return stashUnitProcess;
    }

    private void abc(List<StashUnitTask> parentTasks,StashUnitTask newTask,List<StashUnitTask> taskList,DataStashUnitStep stepDomain){
        parentTasks = new ArrayList<>();
        parentTasks.add(newTask);
        taskList.add(newTask);
        StashUnitStep stashUnitStep = buildStashUnitStep(stepDomain,newTask);
        newTask.getTaskSteps().add(stashUnitStep);
    }
    
    
    public List<StashUnitTask> buildTask(StashUnitDefine unit, StashUnit redisUnit){
    	//build-task-link 
    	//限定： 不允许分支出现
    	Map<String, StashUnitTask> tasks = Maps.newConcurrentMap();
    	
    	List<DataStashUnitStep> rootSteps = unit.getRootSteps();
    	StashUnitTask rootTask = null;
    	if(rootSteps != null && rootSteps.size() > 0){
    		DataStashUnitStep step = rootSteps.get(0);
    		StashUnitTask preTask = null;
    		do{
    			String stepType = step.getStepDefine().getStepType();

    			Preconditions.checkNotNull(stepType);
    			if(stepType.equals(UnitStepTypeEnum.Datax.value()) 
    					|| stepType.equals(UnitStepTypeEnum.DP.value())
    					|| stepType.equals(UnitStepTypeEnum.Sharding.value())){
    				
    				if(preTask == null){
        				preTask = new StashUnitTask(stepType, redisUnit);
        				if(rootTask == null){
        					rootTask = preTask;
        				}
        			} else {
        				if(!preTask.getTaskType().equals(stepType)) {
        					StashUnitTask task = new StashUnitTask(stepType, redisUnit);
            				task.addPreTask(preTask);
            				preTask.addNextTask(task);
            				preTask = task;
        				}
        			}
        			if(!tasks.containsKey(preTask.getTaskProcId())){
        				tasks.put(preTask.getTaskProcId(), preTask);
        			}
        			preTask.addTaskStep(buildStashUnitStep(step, preTask));
    			} 
    			
    			List<DataStashUnitStep> nextSteps = step.getNextSteps();
    			step = null;
    			if(nextSteps != null && nextSteps.size() > 0){
    				step = nextSteps.get(0);
    			}
    		}while(step != null);
    	}
    	
    	//fill-link
    	//sharding 替换处理
    	Preconditions.checkNotNull(rootTask);
    	StashUnitTask curTask = rootTask;
    	
    	StashUnitTask preTask = null;
    	StashUnitTask nextTask = null;
    	do{
    		if(!Strings.isNullOrEmpty(curTask.nextProcId()) ){
    			nextTask = tasks.get(curTask.nextProcId());
    		}else{
    			nextTask = null;
    		}
    		
    		if(curTask.getTaskType().equals(UnitStepTypeEnum.Sharding.value())){ 
    			StashUnitStep step = curTask.getTaskSteps().get(0);
    			DataStashUnitStep stepByInx = unit.getStepByInx(step.getStepIdx());
    			//清理掉前后task对当前task索引
    			if(preTask != null){
    				preTask.cleanNextTask();
    			}
    			if(nextTask != null){
    				nextTask.cleanPreTask();
    			}
    			buildShardingTask(stepByInx, redisUnit, preTask, nextTask, tasks);
    			tasks.remove(curTask.getTaskProcId());
    		}
    		
    		preTask = curTask; 
    		curTask = nextTask;
    	}while(curTask != null);
    	
    	return Lists.newArrayList(tasks.values());
    }
    
    
    public void buildShardingTask(DataStashUnitStep stepByInx, StashUnit redisUnit,
    		StashUnitTask pre, StashUnitTask next, Map<String, StashUnitTask> tasks){
    	
    	List<List<Map<String,String>>> shardContents = 
				unitShardingService.sharding(stepByInx, redisUnit);
    	
    	for(int i = 0; i < shardContents.size();i++){

            String taskType = stepByInx.getChildren().get(0).getStepDefine().getStepType();

            StashUnitTask shardTask = new StashUnitTask(taskType, redisUnit);
            shardTask.setTaskContent(shardContents.get(i));

            tasks.put(shardTask.getTaskProcId(), shardTask);
            for (DataStashUnitStep cStep : stepByInx.getChildren()){
                StashUnitStep stashUnitStep = buildStashUnitStep(cStep, shardTask);
                shardTask.addTaskStep(stashUnitStep);
            } 
            
            if(pre != null){
            	pre.addNextTask(shardTask);
            	shardTask.addPreTask(pre);
            }
            if(next != null){
            	next.addPreTask(shardTask);
            	shardTask.addNextTask(next);
            } 
        }
    }
    
    

    /**
     * 递归步骤，因为步骤是层级结构的
     */
    private void recursiveTask(DataStashUnitStep stepDomain,String parentEngine, List<StashUnitTask> parentTasks, List<StashUnitTask> taskList,StashUnit stashUnit){
        if(UnitStepTypeEnum.Datax.value().equals(stepDomain.getStepDefine().getStepType())
                || UnitStepTypeEnum.DP.value().equals(stepDomain.getStepDefine().getStepType())){
            StashUnitTask newTask = null;
            if(parentEngine==null || !parentEngine.equals(stepDomain.getStepDefine().getStepType())){// 不同的引擎
                parentEngine = stepDomain.getStepDefine().getStepType();
                newTask = new StashUnitTask(stashUnit.getUnitProcId(),stepDomain.getStepDefine().getStepType(),stashUnit.getUnitCode(),stashUnit.getBatchNo());
                if(parentTasks!=null && parentTasks.size()>0){
                    for(StashUnitTask parentTask :parentTasks){
                        newTask.getPreTaskProcIds().add(parentTask.getTaskProcId());
                        parentTask.getNextTaskProcIds().add(newTask.getTaskProcId());
                    }
                }
                abc(parentTasks,newTask,taskList,stepDomain);
                if(parentTasks!=null){
                    parentTasks.clear();
                }else{
                    parentTasks = new ArrayList<>();
                }
                parentTasks.add(newTask);
            }else if(parentTasks!=null && parentTasks.size()>0){
                for(StashUnitTask parentTask :parentTasks){
                    //TODO 这块可能有问题，导致子step的上下级不正确，混编的乱序
                    StashUnitStep stashUnitStep = buildStashUnitStep(stepDomain,parentTask);
                    parentTask.getTaskSteps().add(stashUnitStep);
                }
            }
        }else if(UnitStepTypeEnum.Sharding.value().equals(stepDomain.getStepDefine().getStepType())){

            if(stepDomain.getChildren().size()>0){
                List tmpParentTasks = new ArrayList<>();
                parentEngine = null;

                List<List<Map<String,String>>> shardContents = unitShardingService.sharding(stepDomain,stashUnit);

                for(int i = 0; i < shardContents.size();i++){

                    String taskType = stepDomain.getChildren().get(0).getStepDefine().getStepType();

                    StashUnitTask shardTask = new StashUnitTask(stashUnit.getUnitProcId(),taskType, stashUnit.getUnitCode(), stashUnit.getBatchNo());
                    shardTask.setTaskContent(shardContents.get(i));

                    List<StashUnitStep> taskSteps = new ArrayList<>();

                    for (DataStashUnitStep childDomain : stepDomain.getChildren()){
                        StashUnitStep stashUnitStep = buildStashUnitStep(childDomain,shardTask);
                        taskSteps.add(stashUnitStep);
                    }

                    shardTask.setTaskSteps(taskSteps);

                    if(parentTasks!=null && parentTasks.size()>0){
                        for(StashUnitTask parentTask :parentTasks){
                            shardTask.getPreTaskProcIds().add(parentTask.getTaskProcId());
                            parentTask.getNextTaskProcIds().add(shardTask.getTaskProcId());
                        }
                    }
                    tmpParentTasks.add(shardTask);
                    taskList.add(shardTask);
                }
                parentTasks = tmpParentTasks;
            }else{
                tracelog.error(StashTraceLogVo.buildStashTraceLogVo(stashUnit,clzName,"dispatch","开始分配unitTask"));
            }

        }

        if(stepDomain.getNextSteps()!=null && stepDomain.getNextSteps().size()>0){
            for(DataStashUnitStep childDomain : stepDomain.getNextSteps()){
                recursiveTask(childDomain,parentEngine,parentTasks,taskList,stashUnit);
            }
        }
    }

    private StashUnitStep buildStashUnitStep(DataStashUnitStep stepDomain,StashUnitTask curTask){

        StashUnitStep stashUnitStep = new StashUnitStep();
        stashUnitStep.setStepIdx(stepDomain.getStepIndex());

        if(curTask!=null){
            stashUnitStep.setTaskProcId(curTask.getTaskProcId());
            stashUnitStep.setUnitCode(curTask.getUnitCode());
        }

        stashUnitStep.setNextStepIdxs(stepDomain.getNextStepIdxs());
        stashUnitStep.setPreStepIdxs(stepDomain.getPreStepIdxs());
        stashUnitStep.setParentStepIdx(stepDomain.getParentStepIdx());
        stashUnitStep.setIgnoreErrStepIdxs(stepDomain.getIgnoreErrStepIdxs());
        stashUnitStep.setStepCode(stepDomain.getStepInst());
        stashUnitStep.setStepCfgVal(stepDomain.getStepCfgVal());

        return stashUnitStep;
    }

}
